package com.relytosoft.mizar.analyze.handle.listener;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.mizar.analyze.base.constant.CommonKafkaConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlDataReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlFaceControlReq;
import com.relytosoft.mizar.analyze.handle.service.FaceListenerService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.*;

/**
 * @author wss
 * @version 1.0
 * @date 2022/5/13 17:00
 */
@Component
public class FaceListener {

    @Autowired
    private KafkaProducer kafkaProducer;
    @Autowired
    private FaceListenerService faceListenerService;

    @KafkaListener(topics = FACE_RAW_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleRawData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseObject(r.value().toString(), BtdlDataReq.class))
                .forEach(m -> {
                    if (faceListenerService.handleRawData(m)) {
                        String json = JSON.toJSONString(m);
                        kafkaProducer.send(FACE_IO_DATA_TOPIC, json, true);
                        kafkaProducer.send(FACE_BUSINESS_DATA_TOPIC, json, true);
                    }
                });
        ack.acknowledge();
    }

    @KafkaListener(topics = FACE_IO_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleIOData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseObject(r.value().toString(), BtdlDataReq.class))
                .forEach(m ->
                        faceListenerService.handleIOData(m)
                );
        ack.acknowledge();
    }

    @KafkaListener(topics = FACE_BUSINESS_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleBusinessData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseObject(r.value().toString(), BtdlDataReq.class))
                .forEach(m ->
                        faceListenerService.handleBusinessData(m)
                );
        ack.acknowledge();
    }

    @KafkaListener(topics = FACE_CONTROL_RAW_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleControlRawData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseArray(r.value().toString(), BtdlFaceControlReq.class))
                .forEach(m -> {
                    if (faceListenerService.handleControlRawData(m)) {
                        String json = JSON.toJSONString(m);
                        kafkaProducer.send(FACE_CONTROL_IO_DATA_TOPIC, json, true);
                        kafkaProducer.send(FACE_CONTROL_BUSINESS_DATA_TOPIC, json, true);
                    }
                });
        ack.acknowledge();
    }

    @KafkaListener(topics = FACE_CONTROL_IO_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleControlIoData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseArray(r.value().toString(), BtdlFaceControlReq.class))
                .forEach(m ->
                        faceListenerService.handleControlIOData(m)
                );
        ack.acknowledge();
    }

    @KafkaListener(topics = FACE_CONTROL_BUSINESS_DATA_TOPIC, containerFactory = CommonKafkaConstants.BATCH_CONTAINER_FACTORY)
    public void handleControlBusinessData(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        records
                .stream()
                .map(r -> JSON.parseArray(r.value().toString(), BtdlFaceControlReq.class))
                .forEach(faceControlReqs -> {
                    faceControlReqs.stream().forEach(m ->
                            faceListenerService.handleControlBusinessData(m)
                    );
                });
        ack.acknowledge();
    }

}
